Java UDF

开发Scalar UDF

Scalar UDF，即用户自定义标量函数，可以对单行数据进行操作，输出单行结果。当您在查询时使用Scalar UDF，每行数据最终都会按行出现在结果集中。典型的标量函数包括UPPER、LOWER、ROUND、ABS。

以下示例以提取JSON数据功能为例进行说明。例如，业务场景中，JSON数据中某个字段的值可能是JSON字符串而不是JSON对象，因此在提取JSON字符串时，SQL语句需要嵌套调用GET_JSON_STRING，即GET_JSON_STRING(GET_JSON_STRING('{"key":"{\\"k0\\":\\"v0\\"}"}', "$.key"), "$.k0")。

为简化SQL语句，您可以开发一个UDF，直接提取JSON字符串，例如：MY_UDF_JSON_GET('{"key":"{\\"k0\\":\\"v0\\"}"}', "$.key.k0")。

package com.starrocks.udf.sample;
import com.alibaba.fastjson.JSONPath;

public class UDFJsonGet {
    public final String evaluate(String jsonObj, String key) {
        if (obj == null || key == null) return null;
        try {
            // JSONPath库可以全部展开，即使某个字段的值是JSON格式的字符串
            return JSONPath.read(jsonObj, key).toString();
        } catch (Exception e) {
            return null;
        }
    }
}

用户自定义类必须实现如下方法。

开发UDAF

UDAF，即用户自定义的聚合函数，对多行数据进行操作，输出单行结果。典型的聚合函数包括SUM、COUNT、MAX、MIN，这些函数对于每个GROUP BY分组中多行数据进行聚合后，只输出一行结果。

以下示例以MY_SUM_INT函数为例进行说明。与内置函数SUM（返回值为BIGINT类型）区别在于，MY_SUM_INT函数支持传入参数和返回参数的类型为INT。

package com.starrocks.udf.sample;

public class SumInt {
    public static class State {
        int counter = 0;
        public int serializeLength() { return 4; }
    }

    public State create() {
        return new State();
    }

    public void destroy(State state) {
    }

    public final void update(State state, Integer val) {
        if (val != null) {
            state.counter+= val;
        }
    }

    public void serialize(State state, java.nio.ByteBuffer buff) {
        buff.putInt(state.counter);
    }

    public void merge(State state, java.nio.ByteBuffer buffer) {
        int val = buffer.getInt();
        state.counter += val;
    }

    public Integer finalize(State state) {
        return state.counter;
    }
}

用户自定义类必须实现如下方法。

并且，开发UDAF函数时，您需要使用缓冲区类java.nio.ByteBuffer和局部变量serializeLength，用于保存和表示中间结果，指定中间结果的序列化长度。

开发UDWF

UDWF，即用户自定义窗口函数。跟普通聚合函数不同的是，窗口函数针对一组行（一个窗口）计算值，并为每行返回一个结果。一般情况下，窗口函数包含OVER子句，将数据行拆分成多个分组，窗口函数基于每一行数据所在的组（一个窗口）进行计算，并为每行返回一个结果。

以下示例以MY_WINDOW_SUM_INT函数为例进行说明。与内置函数SUM（返回类型为BIGINT）区别在于，MY_WINDOW_SUM_INT函数支持传入参数和返回参数的类型为INT。

package com.starrocks.udf.sample;

public class WindowSumInt {    
    public static class State {
        int counter = 0;
        public int serializeLength() { return 4; }
        @Override
        public String toString() {
            return "State{" +
                    "counter=" + counter +
                    '}';
        }
    }

    public State create() {
        return new State();
    }

    public void destroy(State state) {

    }

    public void update(State state, Integer val) {
        if (val != null) {
            state.counter+=val;
        }
    }

    public void serialize(State state, java.nio.ByteBuffer buff) {
        buff.putInt(state.counter);
    }

    public void merge(State state, java.nio.ByteBuffer buffer) {
        int val = buffer.getInt();
        state.counter += val;
    }

    public Integer finalize(State state) {
        return state.counter;
    }

    public void reset(State state) {
        state.counter = 0;
    }

    public void windowUpdate(State state,
                            int peer_group_start, int peer_group_end,
                            int frame_start, int frame_end,
                            Integer[] inputs) {
        for (int i = (int)frame_start; i < (int)frame_end; ++i) {
            state.counter += inputs[i];
        }
    }
}

用户自定义类必须实现UDAF所需要的方法（窗口函数是特殊聚合函数）以及windowUpdate()方法。

需要额外实现的方法

开发UDTF

UDTF，即用户自定义表值函数，读入一行数据，输出多个值可视为一张表。表值函数常用于实现行转列。

以下示例以MY_UDF_SPLIT函数为例进行说明。MY_UDF_SPLIT函数支持分隔符为空格，传入参数和返回参数的类型为STRING。

package com.starrocks.udf.sample;

public class UDFSplit{
    public String[] process(String in) {
        if (in == null) return null;
        return in.split(" ");
    }
}

用户自定义类必须实现如下方法。

语法

CREATE [GLOBAL][AGGREGATE | TABLE] FUNCTION function_name(arg_type [, ...])
RETURNS return_type
[PROPERTIES ("key" = "value" [, ...]) ]

参数说明

创建Scalar UDF

执行如下命令，在StarRocks中创建之前示例中的Scalar UDF。

CREATE [GLOBAL] FUNCTION MY_UDF_JSON_GET(string, string) 
RETURNS string
PROPERTIES (
    "symbol" = "com.starrocks.udf.sample.UDFJsonGet", 
    "type" = "StarrocksJar",
    "file" = "http://<YourBucketName>.oss-cn-xxxx-internal.aliyuncs.com/<YourPath>/udf-1.0-SNAPSHOT-jar-with-dependencies.jar"
);

创建UDAF

执行如下命令，在StarRocks中创建之前示例中的UDAF。

CREATE [GLOBAL] AGGREGATE FUNCTION MY_SUM_INT(INT) 
RETURNS INT
PROPERTIES 
( 
    "symbol" = "com.starrocks.udf.sample.SumInt", 
    "type" = "StarrocksJar",
    "file" = "http://<YourBucketName>.oss-cn-xxxx-internal.aliyuncs.com/<YourPath>/udf-1.0-SNAPSHOT-jar-with-dependencies.jar"
);

PROPERTIES里的参数说明与创建Scalar UDF相同。

创建UDWF

执行如下命令，在StarRocks中创建先前示例中的UDWF。

CREATE [GLOBAL] AGGREGATE FUNCTION MY_WINDOW_SUM_INT(Int)
RETURNS Int
PROPERTIES 
(
    "analytic" = "true",
    "symbol" = "com.starrocks.udf.sample.WindowSumInt", 
    "type" = "StarrocksJar", 
    "file" = "http://<YourBucketName>.oss-cn-xxxx-internal.aliyuncs.com/<YourPath>/udf-1.0-SNAPSHOT-jar-with-dependencies.jar"
);

analytic：所创建的函数是否为窗口函数，固定取值为true。其他参数说明与创建Scalar UDF相同。

创建UDTF

执行如下命令，在StarRocks中创建先前示例中的UDTF。

CREATE [GLOBAL] TABLE FUNCTION MY_UDF_SPLIT(string)
RETURNS string
PROPERTIES 
(
    "symbol" = "com.starrocks.udf.sample.UDFSplit", 
    "type" = "StarrocksJar", 
    "file" = "http://<YourBucketName>.oss-cn-xxxx-internal.aliyuncs.com/<YourPath>/udf-1.0-SNAPSHOT-jar-with-dependencies.jar"
);

PROPERTIES里的参数说明与创建Scalar UDF相同。

使用Scalar UDF

执行如下命令，使用步骤六创建的Scalar UDF函数。

SELECT MY_UDF_JSON_GET('{"key":"{\\"in\\":2}"}', '$.key.in');

使用UDAF

执行如下命令，使用步骤六创建的UDAF函数。

SELECT MY_SUM_INT(col1);

使用UDWF

执行如下命令，使用步骤六创建的UDWF函数。

SELECT MY_WINDOW_SUM_INT(intcol) 
            OVER (PARTITION BY intcol2
                  ORDER BY intcol3
                  ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)
FROM test_basic;

使用UDTF

执行如下命令，使用先前示例中的UDTF。

-- 假设存在表 t1，其列 a、b、c1 信息如下。
SELECT t1.a,t1.b,t1.c1 FROM t1;
> output:
1,2.1,"hello world"
2,2.2,"hello UDTF."

-- 使用 MY_UDF_SPLIT() 函数。
SELECT t1.a,t1.b, MY_UDF_SPLIT FROM t1, MY_UDF_SPLIT(t1.c1); 
> output:
1,2.1,"hello"
1,2.1,"world"
2,2.2,"hello"
2,2.2,"UDTF."